package KafkaProducer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.apache.kafka.common.serialization.LongSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.io.*;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class ProducerPiecePy {
    final static private String TOPIC = "kafkaSpark";
    private final static String FILEPATH = "D:\\github\\Kafka-java-example\\trans_file\\producer_file\\englishkfk.txt";
    //private final static String FILEPATH = "english.txt";

    private static Producer<String, String> createProducer() {
        Properties set = new Properties();
        set.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "h-master:9092,h-slave1:9092,h-slave2:9092");
        set.put(ProducerConfig.CLIENT_ID_CONFIG, "ProducerFiles");
        set.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        set.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //set.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
        return new KafkaProducer<>(set);
    }

    private static List<String> readFileCut(File file) throws IOException {
        BufferedReader reader = null;
        List<String> imgList = new ArrayList<>();
        try {
            reader = new BufferedReader(new FileReader(file));
            while (true) {
                String tempString = new String();
                if ((tempString = reader.readLine()) == null) break;
                //imgList.add(tempString);
                String utf8Str = new String(tempString.getBytes("gbk"),"utf-8");
                imgList.add(utf8Str);
            }

            return imgList;
        } catch (FileNotFoundException e) {
            e.printStackTrace();
            return null;
        } catch (IOException e) {
            e.printStackTrace();
            return null;
        } finally {
            if (reader != null) {
                try {
                    reader.close();
                } catch (IOException e1) {
                    e1.printStackTrace();
                }
            }
        }
    }

    private static boolean sendMsg(List<String> fileList, Producer<String, String> producer) {
        String time = (new Date()).toString();
        int i = 1;
        for (String s : fileList) {
            try {
                Thread.sleep(300);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            RecordHeaders headers = new RecordHeaders();
            headers.add("number", Integer.toString(i).getBytes());
            final ProducerRecord<String, String> producerRecord =
                    new ProducerRecord<String, String>(TOPIC, i % 4, time, s, headers);

            try {
                RecordMetadata metadata = producer.send(producerRecord).get();
                System.out.printf("发送文件%d成功！meta(partition=%d, offset=%d) \n", i++
                        , metadata.partition(), metadata.offset());
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
        {
            producer.flush();
            producer.close();
        }
        return false;
    }


    static void runProducer() {
        File file = new File(FILEPATH);
        final Producer<String, String> producer = createProducer();
        try {
            sendMsg(readFileCut(file), producer);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    public static void main(String... args) throws Exception {
        runProducer();
//        File file = new File(FILEPATH);
//        System.out.println(file.getName());
//        System.out.println(readFileCut(file));
    }
}
